package felix.things.router.mqtt.server.broker.cluster;

import cn.hutool.core.codec.Base64;
import com.alibaba.fastjson.JSONObject;
import felix.things.router.mqtt.server.broker.config.BrokerProperties;
import felix.things.router.mqtt.server.broker.internal.InternalMessage;
import felix.things.router.mqtt.server.broker.internal.InternalSendServer;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import static felix.things.router.mqtt.server.store.utils.StoreUtil.getProcessId;

/**
 * @author Felix
 * @date 2020/6/14
 */
@Component
@Slf4j
public class RedisCluster implements MessageListener {
    private static final String CLUSTER_TOPIC = "mqtt-server:cluster";
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private BrokerProperties brokerProperties;
    @Autowired
    private InternalSendServer internalSendServer;

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String content = new String(message.getBody());
        InternalMessage internalMessage = JSONObject.parseObject(content, InternalMessage.class);
        //判断进程ID是否是自身实例,若相同则不发送,否则集群模式下重复发消息
        if (!brokerProperties.getId().equals(internalMessage.getBrokerId()) && !getProcessId("0").equals(internalMessage.getProcessId())) {
            log.debug("收到集群消息: {}", content);
            internalSendServer.sendPublishMessage(internalMessage.getClientId(), internalMessage.getTopic(), MqttQoS.valueOf(internalMessage.getMqttQoS()),
                    Base64.decode(internalMessage.getMessageBase64()), internalMessage.isRetain(), internalMessage.isDup());
        }
    }

    @Async
    public void sendMessage(InternalMessage internalMessage) {
        redisTemplate.convertAndSend(CLUSTER_TOPIC, JSONObject.toJSONString(internalMessage));
    }

}
